package com.zeta.edw.venus

import java.time.format.DateTimeFormatter
import java.time.temporal.ChronoUnit
import java.time.{LocalDateTime, LocalTime}

import org.apache.hadoop.fs.Path

class ExtractHistory extends ExtractData {

  override def beforeProcess(config: Config): Unit = {

    val conf = config.asInstanceOf[ExtractDataConfig]

    val checkStart = conf.checkStart
    val checkEnd = conf.checkEnd

    if (!conf.overwrite) {
      return
    }

    val fs = conf.fileSystem

    val dateTimeFormatter = DateTimeFormatter.ofPattern(conf.datetimeFormat)
    val dateFormat = DateTimeFormatter.ofPattern("yyyyMMdd")

    val startTime = LocalDateTime.parse(checkStart.get, dateTimeFormatter)
    val endTime = LocalDateTime.parse(checkEnd.get, dateTimeFormatter)
    val judgeTime = LocalTime.of(8, 0, 0)

    val startDate = if (startTime.toLocalTime.compareTo(judgeTime) <= 0) {
      startTime.minusDays(1).toLocalDate
    } else {
      startTime.toLocalDate
    }
    val endDate = if (endTime.toLocalTime.compareTo(judgeTime) <= 0) {
      endTime.minusDays(1).toLocalDate
    } else {
      endTime.toLocalDate
    }

    val days = startDate.until(endDate, ChronoUnit.DAYS).toInt

    (0 to days).foreach(day => {
      val dt = startDate.plusDays(day).format(dateFormat)

      if (fs.exists(new Path(s"${conf.hdfsUrl}/user/hive/warehouse/${conf.hiveDatabase}.db/${conf.hiveTable}/dt=$dt"))) {
        fs.delete(new Path(s"${conf.hdfsUrl}/user/hive/warehouse/${conf.hiveDatabase}.db/${conf.hiveTable}/dt=$dt"), true)
      }
    })

  }

  override def beforeJob(config: Config): Unit = {
    val appConf = config.asInstanceOf[ExtractDataConfig]
    if (appConf.checkStart.isEmpty || appConf.checkEnd.isEmpty) {
      throw new Exception("please specify --check-start and --check-end parameters while extract historic data")
    }
  }

  override def rangeStart(config: ExtractDataConfig): String = {
    val dateTimeFormatter = DateTimeFormatter.ofPattern(config.datetimeFormat)

    LocalDateTime.parse(config.checkStart.get, dateTimeFormatter).minusDays(1).format(dateTimeFormatter)
  }

  override def rangeEnd(config: ExtractDataConfig): String = {
    val dateTimeFormatter = DateTimeFormatter.ofPattern(config.datetimeFormat)

    LocalDateTime.parse(config.checkEnd.get, dateTimeFormatter).plusDays(1).format(dateTimeFormatter)
  }
}

object ExtractHistory extends ExtractHistory {
  def main(args: Array[String]): Unit = {
   // startApp(args, new ExtractDataConfig())


  }
}
